package com.hjh.usb.accessory.communication

import com.elvishew.xlog.XLog
import java.io.*
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicBoolean

class AOACommunicatorThreadPool(
    private val inputStreamProvider: () -> InputStream?,
    private val outputStreamProvider: () -> OutputStream?
) {
    companion object {
        private const val BUFFER_SIZE = 64 * 1024
        private const val ACK_TIMEOUT_MS = 60_000L
    }

    private val executor = Executors.newFixedThreadPool(2) // 一个发送，一个接收
    private val running = AtomicBoolean(false)
    private val sending = AtomicBoolean(false)
    private var ackLatch = CountDownLatch(1)

    private var inputStream: InputStream? = null
    private var outputStream: OutputStream? = null

    fun start() {
        inputStream = inputStreamProvider()
        outputStream = outputStreamProvider()
        running.set(true)

        executor.submit {
            try {
                val buffer = ByteArray(BUFFER_SIZE)
                while (running.get()) {
                    val len = inputStream?.read(buffer) ?: break
                    if (len > 0) {
                        val data = buffer.copyOf(len)
                        if (PacketUtils.isAckPacket(data) && sending.get()) {
                            ackLatch.countDown()
                        }
                    }
                    if (firstAckReceived.get()) break
                }
            } catch (e: IOException) {
                XLog.e("接收线程异常: ${e.message}")
                UsbFlow.updateConnectionState(ConnectionState.Error("接收错误: ${e.message}"))
            }
        }
    }

    var firstAckReceived = AtomicBoolean(false)

    fun sendFile(file: File): Boolean {
        if (!running.get()) return false
        if (UsbFlow.usbConnState.value !is ConnectionState.Connected) return false

        executor.submit {
            try {
                val fileSize = file.length()
                var chunkIndex = 0
                val buffer = ByteArray(BUFFER_SIZE) { 0x02 }


                while (true) {
                    val bytesRead = buffer.size // 模拟读取
                    if (bytesRead <= 0) break

                    val packet = PacketUtils.buildDataPacket(
                        fileSize = fileSize,
                        chunkIndex = chunkIndex++,
                        data = buffer,
                        length = bytesRead
                    )

                    sending.set(true)

                    if (!firstAckReceived.get()) {
                        ackLatch = CountDownLatch(1)
                    }

                    try {
                        outputStream?.write(packet)
                        outputStream?.flush()
                        XLog.i("第${chunkIndex}包")
                    } catch (e: IOException) {
                        e.printStackTrace()
                        XLog.e("发送异常: ${e.message}")
                        UsbFlow.updateConnectionState(ConnectionState.Error("发送错误: ${e.message}"))
                        return@submit
                    }

                    // 首包等待 ACK
                    if (!firstAckReceived.get()) {
                        val ackReceived = ackLatch.await(ACK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
                        if (!ackReceived) {
                            UsbFlow.updateConnectionState(ConnectionState.Error("等待 Ack 超时"))
                            return@submit
                        } else {
                            firstAckReceived.set(true)
                        }
                    }

                    sending.set(false)
                }

                XLog.tag("Accessory Send").i("文件发送完成")
            } catch (e: Exception) {
                XLog.e("文件发送异常: ${e.message}")
                UsbFlow.updateConnectionState(ConnectionState.Error("文件发送异常: ${e.message}"))
            }
        }
        return true
    }


    fun sendFileOneByOne(file: File): Boolean {
        if (!running.get()) return false
        if (UsbFlow.usbConnState.value !is ConnectionState.Connected) return false

        executor.submit {
            try {
                val fileSize = file.length()
                var chunkIndex = 0

                // file.inputStream().use { fis ->
                val buffer = ByteArray(BUFFER_SIZE) { 0x02 }

                while (true) {
                    // val bytesRead = fis.read(buffer)
                    val bytesRead = buffer.size
                    if (bytesRead <= 0) break

                    val packet = PacketUtils.buildDataPacket(
                        fileSize = fileSize,
                        chunkIndex = chunkIndex++,
                        data = buffer,
                        length = bytesRead
                    )

                    sending.set(true)
                    ackLatch = CountDownLatch(1)

                    try {
                        outputStream?.write(packet)
                        outputStream?.flush()
                    } catch (e: IOException) {
                        UsbFlow.updateConnectionState(ConnectionState.Error("发送错误: ${e.message}"))
                        return@submit
                    }

                    val ackReceived = ackLatch.await(ACK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
                    if (!ackReceived) {
                        UsbFlow.updateConnectionState(ConnectionState.Error("等待 Ack 超时"))
                        return@submit
                    }

                    sending.set(false)
                }
                // }

                XLog.tag("Accessory Send").i("文件发送完成")
            } catch (e: Exception) {
                XLog.e("文件发送异常: ${e.message}")
                UsbFlow.updateConnectionState(ConnectionState.Error("文件发送异常: ${e.message}"))
            }
        }

        return true
    }

    fun stop() {
        running.set(false)

        // 提前解除阻塞，防止 await 卡住
        ackLatch.countDown()

        try {
            inputStream?.close()
            outputStream?.close()
        } catch (_: Exception) {
        }

        executor.shutdownNow()

        XLog.tag("Accessory").i("线程池通信已停止，资源释放完毕")
    }
}
